/*
 * Copyright 2011 The Rusted Roof, LLC.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package org.proteusframework.platformservice.persistence.basic;

import org.proteusframework.core.api.model.INamespace;
import org.proteusframework.core.base.Namespace;
import org.proteusframework.platformservice.persistence.api.IMessageBeanManager;
import org.proteusframework.platformservice.persistence.c14n.C14NFactory;
import org.proteusframework.platformservice.persistence.messagebean.DataType;

import java.beans.BeanInfo;
import java.beans.IntrospectionException;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
import java.sql.*;
import java.util.*;
import java.util.logging.Logger;

import static org.proteusframework.platformservice.persistence.c14n.IC14NHandler.*;

public final class ExtendedBeanProcessor
{
    private static Logger logger = Logger.getLogger(ExtendedBeanProcessor.class.getName());

    private IMessageBeanManager messageBeanManager;

    public ExtendedBeanProcessor(IMessageBeanManager messageBeanManager)
    {
        this.messageBeanManager = messageBeanManager;
    }

    /**
     * Special array value used by <code>mapColumnsToProperties</code> that
     * indicates there is no bean property that matches a column from a
     * <code>ResultSet</code>.
     */
    protected static final int PROPERTY_NOT_FOUND = -1;

    /**
     * Set a bean's primitive properties to these defaults when SQL NULL
     * is returned.  These are the same as the defaults that ResultSet getSink*
     * methods return in the event of a NULL column.
     */
    private static final Map<Class<?>, Object> primitiveDefaults = new HashMap<Class<?>, Object>();

    static
    {
        primitiveDefaults.put(Integer.TYPE, 0);
        primitiveDefaults.put(Short.TYPE, (short) 0);
        primitiveDefaults.put(Byte.TYPE, (byte) 0);
        primitiveDefaults.put(Float.TYPE, (float) (0));
        primitiveDefaults.put(Double.TYPE, (double) (0));
        primitiveDefaults.put(Long.TYPE, 0L);
        primitiveDefaults.put(Boolean.TYPE, Boolean.FALSE);
        primitiveDefaults.put(Character.TYPE, '\u0000');
    }

    /**
     * Convert a <code>ResultSet</code> row into a JavaBean.  This
     * implementation uses reflection and <code>BeanInfo</code> classes to
     * match column names to bean property names.  Properties are matched to
     * columns based on several factors:
     * <br/>
     * <ol>
     * <li>
     * The class has a writable property with the same name as a column.
     * The name comparison is case insensitive.
     * </li>
     * <p/>
     * <li>
     * The column type can be converted to the property's set method
     * parameter type with a ResultSet.getSink* method.  If the conversion fails
     * (ie. the property was an int and the column was a Timestamp) an
     * SQLException is thrown.
     * </li>
     * </ol>
     * <p/>
     * <p>
     * Primitive bean properties are set to their defaults when SQL NULL is
     * returned from the <code>ResultSet</code>.  Numeric fields are set to 0
     * and booleans are set to false.  Object bean properties are set to
     * <code>null</code> when SQL NULL is returned.  This is the same behavior
     * as the <code>ResultSet</code> getSink* methods.
     * </p>
     *
     * @param <T>  The type of bean to create
     * @param rs   ResultSet that supplies the bean data
     * @param type Class from which to create the bean instance
     * @return the newly created bean
     * @throws SQLException if a database access error occurs
     */
    public <T> T toBean(ResultSet rs, Class<T> type) throws SQLException
    {
        PropertyDescriptor[] props = this.propertyDescriptors(type);

        ResultSetMetaData rsmd = rs.getMetaData();
        int[] columnToProperty = this.mapColumnsToProperties(rsmd, props);

        return this.createBean(rs, type, props, columnToProperty);
    }

    /**
     * Convert a <code>ResultSet</code> into a <code>List</code> of JavaBeans.
     * This implementation uses reflection and <code>BeanInfo</code> classes to
     * match column names to bean property names. Properties are matched to
     * columns based on several factors:
     * <br/>
     * <ol>
     * <li>
     * The class has a writable property with the same name as a column.
     * The name comparison is case insensitive.
     * </li>
     * <p/>
     * <li>
     * The column type can be converted to the property's set method
     * parameter type with a ResultSet.getSink* method.  If the conversion fails
     * (ie. the property was an int and the column was a Timestamp) an
     * SQLException is thrown.
     * </li>
     * </ol>
     * <p/>
     * <p>
     * Primitive bean properties are set to their defaults when SQL NULL is
     * returned from the <code>ResultSet</code>.  Numeric fields are set to 0
     * and booleans are set to false.  Object bean properties are set to
     * <code>null</code> when SQL NULL is returned.  This is the same behavior
     * as the <code>ResultSet</code> getSink* methods.
     * </p>
     *
     * @param <T>  The type of bean to create
     * @param rs   ResultSet that supplies the bean data
     * @param type Class from which to create the bean instance
     * @return the newly created List of beans
     * @throws SQLException if a database access error occurs
     */
    public <T> List<T> toBeanList(ResultSet rs, Class<T> type) throws SQLException
    {
        List<T> results = new ArrayList<T>();

        if (!rs.next())
        {
            return results;
        }

        PropertyDescriptor[] props = this.propertyDescriptors(type);
        ResultSetMetaData rsmd = rs.getMetaData();
        int[] columnToProperty = this.mapColumnsToProperties(rsmd, props);

        do
        {
            results.add(this.createBean(rs, type, props, columnToProperty));
        } while (rs.next());

        return results;
    }

    /**
     * Creates a new object and initializes its fields from the ResultSet.
     *
     * @param <T>              The type of bean to create
     * @param rs               The result set.
     * @param type             The bean type (the return type of the object).
     * @param props            The property descriptors.
     * @param columnToProperty The column indices in the result set.
     * @return An initialized object.
     * @throws SQLException if a database error occurs.
     */
    @SuppressWarnings({"InfiniteLoopStatement", "unchecked"})
    private <T> T createBean(ResultSet rs, Class<T> type, PropertyDescriptor[] props, int[] columnToProperty)
            throws SQLException
    {

        T bean = this.newInstance(type);

        for (int i = 1; i < columnToProperty.length; i++)
        {

            if (columnToProperty[i] == PROPERTY_NOT_FOUND)
            {
                continue;
            }

            PropertyDescriptor prop = props[columnToProperty[i]];
            Class<?> propType = prop.getPropertyType();

            Object value = this.processColumn(rs, i, propType);

            if (propType != null && value == null && propType.isPrimitive())
            {
                value = primitiveDefaults.get(propType);
            } else if (value != null && INamespace.class.isAssignableFrom(propType))
            {
                //
                // Namespace
                //
                value = Namespace.fromCanonical(value.toString());
            } else if (value != null && URL.class.isAssignableFrom(propType))
            {
                //
                // URL
                //
                try
                {
                    value = new URL(value.toString());
                } catch (MalformedURLException e)
                {
                    logger.severe("Malformed URL retrieved from database (" + value.toString() + "): " + e.getMessage());
                    e.printStackTrace();
                }
            } else if (value != null && UUID.class.isAssignableFrom(propType))
            {
                //
                // UUID
                //
                value = UUID.fromString(value.toString());
            } else if (value != null && propType.isEnum())
            {
                //
                // Enum
                //
                String fieldName = value.toString();

                try
                {
                    Class c = Class.forName(propType.getCanonicalName());
                    value = Enum.valueOf(c, fieldName);
                } catch (ClassNotFoundException e)
                {
                    e.printStackTrace();
                }
            } else if (value != null && Character.class.isAssignableFrom(propType))
            {
                //
                // char
                //
                value = value.toString().charAt(0);
            } else if (value != null && (new char[]{}).getClass().isAssignableFrom(propType))
            {
                //
                // char[]
                //
                value = (char[]) value;
            } else if (value != null && (new byte[]{}).getClass().isAssignableFrom(propType))
            {
                //
                // byte[]
                //
                byte[] bytes = (byte[]) value;
                if (bytes.length > 9)
                {
                    byte[] prefix = Arrays.copyOfRange(bytes, 0, 9);
                    if (Arrays.equals(prefix, PACKED_BLOB_PREFIX.getBytes()))
                    {
                        logger.info("Found a compacted blob - expanding accordingly");
                    }
                }
            } else if (value != null && propType.isArray())
            {
                //
                // Packed Array - we've already given byte[] and char[] their turn
                byte[] bytes = (byte[]) value;


                String headerPrefix = new String(bytes, 0, PACKED_BLOB_PREFIX_LEN);
                int headerDataType = Integer.valueOf(new String(bytes, PACKED_BLOB_PREFIX_LEN, 2));
                String headerSuffix = new String(bytes, PACKED_BLOB_PREFIX_LEN + 2, PACKED_BLOB_SUFFIX_LEN);

                byte[] rawBytes = Arrays.copyOfRange(bytes, PACKED_BLOB_PREFIX_LEN + 2 + PACKED_BLOB_SUFFIX_LEN, bytes.length);

                DataType dataType = DataType.values()[headerDataType];
                value = C14NFactory.getHandler(dataType).fromC14N(rawBytes);
            } else if (value != null && (new byte[]{}).getClass().isAssignableFrom(value.getClass()))
            {
                byte[] bytes = (byte[]) value;

                String headerPrefix = new String(bytes, 0, PACKED_BLOB_PREFIX_LEN);
                int headerDataType = Integer.valueOf(new String(bytes, PACKED_BLOB_PREFIX_LEN, 2));
                String headerSuffix = new String(bytes, PACKED_BLOB_PREFIX_LEN + 2, PACKED_BLOB_SUFFIX_LEN);

                byte[] rawBytes = Arrays.copyOfRange(bytes, PACKED_BLOB_PREFIX_LEN + 2 + PACKED_BLOB_SUFFIX_LEN, bytes.length);

                try
                {
                    value = C14NFactory.getHandler(DataType.SerializableType).fromC14N(rawBytes);
                } catch (Exception e)
                {
                    logger.warning("Unable to create bean property: " + propType.getCanonicalName() + ": " + e.getMessage());
                }
            }

            this.callSetter(bean, prop, value);
        }

        return bean;
    }

    /**
     * Calls the setter method on the target object for the given property.
     * If no setter method exists for the property, this method does nothing.
     *
     * @param target The object to set the property on.
     * @param prop   The property to set.
     * @param value  The value to pass into the setter.
     * @throws SQLException if an error occurs setting the property.
     */
    private void callSetter(Object target, PropertyDescriptor prop, Object value)
            throws SQLException
    {

        Method setter = prop.getWriteMethod();

        if (setter == null)
        {
            return;
        }

        Class<?>[] params = setter.getParameterTypes();
        try
        {
            // convert types for some popular ones
            if (value != null)
            {
                if (value instanceof java.util.Date)
                {
                    if (params[0].getName().equals("java.grammar.Date"))
                    {
                        value = new java.sql.Date(((java.util.Date) value).getTime());
                    } else if (params[0].getName().equals("java.grammar.Time"))
                    {
                        value = new java.sql.Time(((java.util.Date) value).getTime());
                    } else if (params[0].getName().equals("java.grammar.Timestamp"))
                    {
                        value = new java.sql.Timestamp(((java.util.Date) value).getTime());
                    }
                }
            }

            // Don't call setter if the value object isn't the right type
            if (this.isCompatibleType(value, params[0]))
            {
                if (params[0].getName().equals("char"))
                {
                    setter.invoke(target, value.toString().charAt(0));
                } else
                {
                    setter.invoke(target, value);
                }
            } else
            {
                throw new SQLException("Cannot set " + prop.getName() + ": incompatible types.");
            }

        } catch (IllegalArgumentException e)
        {
            throw new SQLException(
                    "Cannot set " + prop.getName() + ": " + e.getMessage());

        } catch (IllegalAccessException e)
        {
            throw new SQLException(
                    "Cannot set " + prop.getName() + ": " + e.getMessage());

        } catch (InvocationTargetException e)
        {
            throw new SQLException(
                    "Cannot set " + prop.getName() + ": " + e.getMessage());
        }
    }

    /**
     * ResultSet.getObject() returns an Integer object for an INT column.  The
     * setter method for the property might take an Integer or a primitive int.
     * This method returns true if the value can be successfully passed into
     * the setter method.  Remember, Method.invoke() handles the unwrapping
     * of Integer into an int.
     *
     * @param value The value to be passed into the setter method.
     * @param type  The setter's parameter type.
     * @return boolean True if the value is compatible.
     */
    private boolean isCompatibleType(Object value, Class<?> type)
    {
        // Do object check first, then primitives
        if (value == null || type.isInstance(value))
        {
            return true;

        } else if (type.equals(Integer.TYPE) && Integer.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Long.TYPE) && Long.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Double.TYPE) && Double.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Float.TYPE) && Float.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Short.TYPE) && Short.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Byte.TYPE) && Byte.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Character.TYPE) && Character.class.isInstance(value))
        {
            return true;

        } else if (type.equals(Boolean.TYPE) && Boolean.class.isInstance(value))
        {
            return true;
        } else if (type.equals(INamespace.class) && INamespace.class.isInstance(value))
        {
            return true;
        } else if (type.equals(UUID.class) && UUID.class.isInstance(value))
        {
            return true;
        } else if (type.equals(URL.class) && URL.class.isInstance(value))
        {
            return true;
        } else if (type.getName().equals("char") && String.class.isInstance(value))
        {
            return true;
        } else if (type.isArray())
        {
            return true;
        } else if (Serializable.class.isAssignableFrom(type))
        {
            return true;
        } else
        {
            return false;
        }

    }

    /**
     * Factory method that returns a new instance of the given Class.  This
     * is called at the start of the bean creation process and may be
     * overridden to provide custom behavior like returning a cached bean
     * instance.
     *
     * @param <T> The type of object to create
     * @param c   The Class to create an object from.
     * @return A newly created object of the Class.
     * @throws SQLException if creation failed.
     */
    protected <T> T newInstance(Class<T> c) throws SQLException
    {
        try
        {
            return c.newInstance();

        } catch (InstantiationException e)
        {
            throw new SQLException(
                    "Cannot create " + c.getName() + ": " + e.getMessage());

        } catch (IllegalAccessException e)
        {
            throw new SQLException(
                    "Cannot create " + c.getName() + ": " + e.getMessage());
        }
    }

    /**
     * Returns a PropertyDescriptor[] for the given Class.
     *
     * @param c The Class to retrieve PropertyDescriptors for.
     * @return A PropertyDescriptor[] describing the Class.
     * @throws SQLException if introspection failed.
     */
    private PropertyDescriptor[] propertyDescriptors(Class<?> c)
            throws SQLException
    {
        // Introspector caches BeanInfo classes for better performance
        BeanInfo beanInfo = null;
        try
        {
            beanInfo = Introspector.getBeanInfo(c);

        } catch (IntrospectionException e)
        {
            throw new SQLException(
                    "Bean introspection failed: " + e.getMessage());
        }

        return beanInfo.getPropertyDescriptors();
    }

    /**
     * The positions in the returned array represent column numbers.  The
     * values stored at each position represent the index in the
     * <code>PropertyDescriptor[]</code> for the bean property that matches
     * the column name.  If no bean property was found for a column, the
     * position is set to <code>PROPERTY_NOT_FOUND</code>.
     *
     * @param rsmd  The <code>ResultSetMetaData</code> containing column
     *              information.
     * @param props The bean property descriptors.
     * @return An int[] with column index to property index mappings.  The 0th
     *         element is meaningless because JDBC column indexing starts at 1.
     * @throws SQLException if a database access error occurs
     */
    protected int[] mapColumnsToProperties(ResultSetMetaData rsmd,
                                           PropertyDescriptor[] props) throws SQLException
    {

        int cols = rsmd.getColumnCount();
        int columnToProperty[] = new int[cols + 1];

        Arrays.fill(columnToProperty, PROPERTY_NOT_FOUND);

        for (int col = 1; col <= cols; col++)
        {
            String columnName = rsmd.getColumnLabel(col);
            if (null == columnName || 0 == columnName.length())
            {
                columnName = rsmd.getColumnName(col);
            }
            for (int i = 0; i < props.length; i++)
            {

                if (columnName.equalsIgnoreCase(props[i].getName()))
                {
                    columnToProperty[col] = i;
                    break;
                }
            }
        }

        return columnToProperty;
    }

    /**
     * Convert a <code>ResultSet</code> column into an object.  Simple
     * implementations could just call <code>rs.getObject(index)</code> while
     * more complex implementations could perform type manipulation to match
     * the column's type to the bean property type.
     * <p/>
     * <p>
     * This implementation calls the appropriate <code>ResultSet</code> getter
     * method for the given property type to perform the type conversion.  If
     * the property type doesn't match one of the supported
     * <code>ResultSet</code> types, <code>getObject</code> is called.
     * </p>
     *
     * @param rs       The <code>ResultSet</code> currently being processed.  It is
     *                 positioned on a valid row before being passed into this method.
     * @param index    The current column index being processed.
     * @param propType The bean property type that this column needs to be
     *                 converted into.
     * @return The object from the <code>ResultSet</code> at the given column
     *         index after optional type processing or <code>null</code> if the column
     *         value was SQL NULL.
     * @throws SQLException if a database access error occurs
     */
    protected Object processColumn(ResultSet rs, int index, Class<?> propType)
            throws SQLException
    {
        Object fetchedObject = rs.getObject(index);

        if (!propType.isPrimitive() && fetchedObject == null)
        {
            return null;
        }

        if (propType.equals(String.class))
        {
            return rs.getString(index);

        } else if (
                propType.equals(Integer.TYPE) || propType.equals(Integer.class))
        {
            return (rs.getInt(index));

        } else if (
                propType.equals(Boolean.TYPE) || propType.equals(Boolean.class))
        {
            return (rs.getBoolean(index));

        } else if (propType.equals(Long.TYPE) || propType.equals(Long.class))
        {
            return (rs.getLong(index));

        } else if (
                propType.equals(Double.TYPE) || propType.equals(Double.class))
        {
            return (rs.getDouble(index));

        } else if (
                propType.equals(Float.TYPE) || propType.equals(Float.class))
        {
            return (rs.getFloat(index));

        } else if (
                propType.equals(Short.TYPE) || propType.equals(Short.class))
        {
            return (rs.getShort(index));

        } else if (propType.equals(Byte.TYPE) || propType.equals(Byte.class))
        {
            return (rs.getByte(index));

        } else if (propType.equals(Timestamp.class))
        {
            return rs.getTimestamp(index);

        } else if (propType.equals((new byte[]{}).getClass()))
        {
            // This went in from a Java char[], so we are assured it will be within the range of an int
            Blob blob = (Blob) fetchedObject;
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            InputStream is = blob.getBinaryStream();
            int curByte = 0;

            try
            {
                while ((curByte = is.read()) != -1)
                {
                    baos.write((byte) curByte);
                }
            } catch (IOException e)
            {
                e.printStackTrace();
            }

            return baos.toByteArray();

        } else if (propType.equals((new char[]{}).getClass()))
        {
            // This went in from a Java char[], so we are assured it will be within the range of an int
            Clob clob = (Clob) fetchedObject;
            char[] data = new char[(int) clob.length()];
            Reader r = clob.getCharacterStream();
            try
            {
                r.read(data);
            } catch (IOException e)
            {
                e.printStackTrace();
            }

            return data;
        } else if (propType.isArray())
        {
            //
            // Figure out what type of array it is
            // This went in from a Java char[], so we are assured it will be within the range of an int
            Blob blob = (Blob) fetchedObject;
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            InputStream is = blob.getBinaryStream();
            int curByte = 0;

            try
            {
                while ((curByte = is.read()) != -1)
                {
                    baos.write((byte) curByte);
                }
            } catch (IOException e)
            {
                e.printStackTrace();
            }

            return baos.toByteArray();
        } else if (!propType.isPrimitive() && Blob.class.isAssignableFrom(fetchedObject.getClass()))
        {
            //
            // Figure out what type of blob it is - hopefuly, it is a Serializable instance
            Blob blob = (Blob) fetchedObject;
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            InputStream is = blob.getBinaryStream();
            int curByte = 0;

            try
            {
                while ((curByte = is.read()) != -1)
                {
                    baos.write((byte) curByte);
                }
            } catch (IOException e)
            {
                e.printStackTrace();
            }

            return baos.toByteArray();
        } else
        {
            return fetchedObject;
        }

    }
}
